package cn.itcast.tags.models.statistics

import cn.itcast.tags.models.{AbstractModel, ModelType}
import cn.itcast.tags.tools.TagTools
import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions._
class PayTypeModel extends AbstractModel("支付方式",ModelType.STATISTICS){
  override def doTag(businessDF: DataFrame, tagDF: DataFrame): DataFrame = {
    import businessDF.sparkSession.implicits._
    businessDF.show(10,false)
    tagDF.filter($"level" === 5).show(5,false)

    //1.从业务数据中计算获取每个用户使用最多的支付方式
    val paymentDF: DataFrame = businessDF
      //先按照用户ID分组，再按照支付编码paymentcode分组，使用count函数统计次数
      .groupBy($"memberid", $"paymentcode")
      .count() //使用count函数列名就是count
      //使用窗口分析函数获取最多次数
      .withColumn(
        "rnk",
        row_number().over(
          Window.partitionBy($"memberid").orderBy($"count".desc)
        )
      )
      .where($"rnk" === 1) //只取支付方式最多的支付方式
      .select($"memberid".as("id"), $"paymentcode".as("payment"))


    //paymentDF.show(100,false)

    val modelDF: DataFrame = TagTools.ruleMatchTag(paymentDF, "payment", tagDF)
    //modelDF.show(1000,false)
    modelDF
  }
}
object PayTypeModel{
  def main(args: Array[String]): Unit = {
    val model = new PayTypeModel
    model.executeModel(356L,false)
  }
}